package com.vict.task.core.impl;

import cn.hutool.core.collection.ConcurrentHashSet;
import cn.hutool.core.util.IdUtil;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.vict.task.annotation.LocalTaskConsumerConfiguration;
import com.vict.task.bean.dto.IdTimeDTO;
import com.vict.task.bean.dto.LocalTask;
import com.vict.task.bean.dto.LocalTaskConsumeResult;
import com.vict.task.bean.entity.LocalTaskExecuteInstance;
import com.vict.task.config.LocalTaskCommon;
import com.vict.task.core.LocalTaskExecute;
import com.vict.task.core.LocalTaskService;
import com.vict.task.dao.LocalTaskExecuteInstanceDao;
import com.vict.task.utils.ObjectUtil;
import com.vict.task.utils.ThreadUtil;
import com.vict.task.utils.lock.LocalTaskLockApi;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager;

import javax.annotation.PostConstruct;
import java.lang.reflect.Array;
import java.sql.Timestamp;
import java.util.*;
import java.util.concurrent.*;

/**
 * version 1.2
 * 重试机制
 */
@Slf4j
@Component
public class LocalTaskExecuteProcessor {
    public static ExecutorService TestExecuteProcessorThreadPoolDeamon = null;
    public static ExecutorService[] TestExecuteProcessorThreadPoolWork = null;
    public static ExecutorService TestExecuteProcessorThreadPoolAction = new ThreadPoolExecutor(100, 200, 99999, TimeUnit.DAYS, new ArrayBlockingQueue<>(10000));

    public static ExecutorService syncThreadPool = new ThreadPoolExecutor(100, 200, 99999, TimeUnit.DAYS, new ArrayBlockingQueue<>(10000));

    public static ConcurrentHashMap<IdTimeDTO, IdTimeDTO> idTimeMap = new ConcurrentHashMap<IdTimeDTO, IdTimeDTO>();

    @Autowired(required = false)
    List<LocalTaskExecute> localTaskExecutes;

    @Autowired
    LocalTaskExecuteInstanceDao localTaskExecuteInstanceDao;

    @Autowired
    LocalTaskService localTaskService;

    @Autowired
    LocalTaskExecuteProcessor localTaskExecuteProcessor;

    @PostConstruct
    public void init() {
        log.info("本地任务执行器, 当前注册执行器数量:{}", Optional.ofNullable(localTaskExecutes).map(List::size).orElse(0));

        Integer taskNum = Optional.ofNullable(localTaskExecutes).map(List::size).orElse(0);
        if(taskNum == 0){
            return;
        }

        // 执行器守护线程
        LocalTaskExecuteProcessor.TestExecuteProcessorThreadPoolDeamon = new ThreadPoolExecutor(taskNum, taskNum, 99999, TimeUnit.DAYS, new ArrayBlockingQueue<>(taskNum));
        // 执行器工作线程
        TestExecuteProcessorThreadPoolWork = (ExecutorService[]) Array.newInstance(ExecutorService.class, taskNum);
        for(int i = 0; i < localTaskExecutes.size() ; i++){
            String type = Optional.ofNullable(localTaskExecutes.get(i)).map(o -> o.getClass())
                    .map(o -> o.getAnnotation(LocalTaskConsumerConfiguration.class))
                    .map(LocalTaskConsumerConfiguration::type).orElseThrow(() -> new RuntimeException("本地任务执行器TaskConsumerConfiguration->type不可为空"));
            int sameTimeWorkNum = Optional.ofNullable(localTaskExecutes.get(i)).map(o -> o.getClass())
                    .map(o -> o.getAnnotation(LocalTaskConsumerConfiguration.class))
                    .map(LocalTaskConsumerConfiguration::sameTimeWorkNum).get();

            TestExecuteProcessorThreadPoolWork[i] = new ThreadPoolExecutor(sameTimeWorkNum, sameTimeWorkNum, 99999, TimeUnit.DAYS, new ArrayBlockingQueue<>(100000));
        }
        // 启动守护线程
        for(int i = 0; i < localTaskExecutes.size() ; i++){
            LocalTaskExecute localTaskExecute = localTaskExecutes.get(i);
            int finalI = i;
            LocalTaskExecuteProcessor.TestExecuteProcessorThreadPoolDeamon.submit(()-> {
                while(true){
                    try{
                        Thread.sleep(LocalTaskCommon.taskScanTime);
                        String type = Optional.ofNullable(localTaskExecute).map(o -> o.getClass())
                                .map(o -> o.getAnnotation(LocalTaskConsumerConfiguration.class))
                                .map(LocalTaskConsumerConfiguration::type).orElseThrow(() -> new RuntimeException("本地任务执行器TaskConsumerConfiguration->type不可为空"));
                        Integer sameTimeWorkNum = Optional.of(localTaskExecute).map(o -> o.getClass())
                                .map(o -> o.getAnnotation(LocalTaskConsumerConfiguration.class))
                                .map(LocalTaskConsumerConfiguration::sameTimeWorkNum).orElse(null);
                        int[] failRetryTimeSeconds = Optional.ofNullable(localTaskExecute).map(o -> o.getClass())
                                .map(o -> o.getAnnotation(LocalTaskConsumerConfiguration.class))
                                .map(LocalTaskConsumerConfiguration::failRetryTimeSeconds).get();

                        startTaskExecuteProcessor(finalI, localTaskExecute, type, sameTimeWorkNum, failRetryTimeSeconds);
                    }catch(Exception e){
                        log.error("本地任务执行器, 执行异常", e);
                    }
                }
            });
        }
    }


    /**
     * 直接触发
     */
    public void startTaskExecuteProcessor(String type){
        Integer index = getTaskExecuteByType(type);
        if(index == null){
            return;
        }
        TestExecuteProcessorThreadPoolAction.submit(()-> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException ignored) { }

            Integer sameTimeWorkNum = Optional.ofNullable(localTaskExecutes.get(index)).map(o -> o.getClass())
                    .map(o -> o.getAnnotation(LocalTaskConsumerConfiguration.class))
                    .map(LocalTaskConsumerConfiguration::sameTimeWorkNum).get();
            int[] failRetryTimeSeconds = Optional.ofNullable(localTaskExecutes.get(index)).map(o -> o.getClass())
                    .map(o -> o.getAnnotation(LocalTaskConsumerConfiguration.class))
                    .map(LocalTaskConsumerConfiguration::failRetryTimeSeconds).get();

            startTaskExecuteProcessor(index, localTaskExecutes.get(index), type, sameTimeWorkNum, failRetryTimeSeconds);
            return null;
        });
    }

    /**
     * 直接触发同步
     */
    public void startTaskExecuteProcessorSync(String type, LocalTaskExecuteInstance localTaskExecuteInstance){
        Integer index = getTaskExecuteByType(type);
        if(index == null){
            return;
        }

        LocalTaskExecute localTaskExecute = localTaskExecutes.get(index);

        int[] failRetryTimeSeconds = Optional.ofNullable(localTaskExecutes.get(index)).map(o -> o.getClass())
                .map(o -> o.getAnnotation(LocalTaskConsumerConfiguration.class))
                .map(LocalTaskConsumerConfiguration::failRetryTimeSeconds).get();

        String id = localTaskExecuteInstance.getTaskExecuteInstanceID();
        LocalTaskLockApi lockMain = LocalTaskLockApi.getLock("localTaskLockMain" + "-id-" + id);

        if(lockMain.lockNotBlock()){
            try{
                startTaskExecuteProcessorOneSync(localTaskExecute, localTaskExecuteInstance, failRetryTimeSeconds);
            }finally{
                // 执行完毕, 锁进行释放
                lockMain.unlockIfSuccess();
            }
        }else{
            log.info("已经有任务正在进行, type={}, taskExecuteInstance={}", type, ObjectUtil.toString(localTaskExecuteInstance));
        }
    }

    private Integer getTaskExecuteByType(String type){
        for(int i = 0; i < localTaskExecutes.size() ; i++){
            LocalTaskExecute localTaskExecute = localTaskExecutes.get(i);
            Boolean b = Optional.ofNullable(localTaskExecute).map(o -> o.getClass())
                    .map(o -> o.getAnnotation(LocalTaskConsumerConfiguration.class))
                    .map(LocalTaskConsumerConfiguration::type).map(o -> o.equals(type)).orElse(false);
            if(b){
                return i;
            }
        }
        return null;
    }

    public void startTaskExecuteProcessor(int index, LocalTaskExecute localTaskExecute, String type, Integer sameTimeWorkNum, int[] failRetryTimeSeconds){
        // 查找任务
        List<LocalTaskExecuteInstance> localTaskExecuteInstances = selectTaskList(type, sameTimeWorkNum);
        if(localTaskExecuteInstances == null || localTaskExecuteInstances.size() == 0){
            log.debug("任务扫描无数据 now={}", System.currentTimeMillis());
            return;
        }

        for(LocalTaskExecuteInstance localTaskExecuteInstance : localTaskExecuteInstances){
            TestExecuteProcessorThreadPoolWork[index].submit(()-> {
                startTaskExecuteProcessorOne(localTaskExecuteInstance, localTaskExecute, failRetryTimeSeconds);
                return null;
            });
        }
    }

    private void startTaskExecuteProcessorOne(LocalTaskExecuteInstance localTaskExecuteInstance, LocalTaskExecute localTaskExecute, int[] failRetryTimeSeconds){
        IdTimeDTO idTimeDTO = new IdTimeDTO();
        idTimeDTO.setId(localTaskExecuteInstance.getTaskExecuteInstanceID());
        idTimeDTO.setActionTime(localTaskExecuteInstance.getActionTime().getTime());
        idTimeDTO.setOverTimestamp(System.currentTimeMillis() + 1000 * 60 * 5);

        IdTimeDTO idTimeDTOOld = idTimeMap.putIfAbsent(idTimeDTO, idTimeDTO);
        if(idTimeDTOOld == null){ // 添加成功
            // 任务发起成功
            log.info("任务发起成功, id={}", localTaskExecuteInstance.getTaskExecuteInstanceID());
            startTaskExecuteProcessorOneMain(localTaskExecuteInstance, localTaskExecute, failRetryTimeSeconds);
        }else{
            // 已经存在
            long now = System.currentTimeMillis();
            if(idTimeDTOOld.getOverTimestamp() <= now){
                // 过期
                idTimeMap.remove(idTimeDTOOld); // 移除旧任务
                idTimeDTOOld = idTimeMap.putIfAbsent(idTimeDTO, idTimeDTO); // 添加新任务
                if(idTimeDTOOld == null){ // 添加成功
                    log.info("任务过期 再次发起成功, id={}", localTaskExecuteInstance.getTaskExecuteInstanceID());
                    startTaskExecuteProcessorOneMain(localTaskExecuteInstance, localTaskExecute, failRetryTimeSeconds);
                }else{
                    log.info("任务过期 再次发起失败, 无需处理, id={}", localTaskExecuteInstance.getTaskExecuteInstanceID());
                }
            }else{
                // 未过期
                log.info("任务未过期 无需发起, id={}", localTaskExecuteInstance.getTaskExecuteInstanceID());
            }
        }
    }

    private void startTaskExecuteProcessorOneMain(LocalTaskExecuteInstance localTaskExecuteInstance, LocalTaskExecute localTaskExecute, int[] failRetryTimeSeconds){
        Date date = new Date(localTaskExecuteInstance.getActionTime().getTime());
        Timer timer = new Timer();
        timer.schedule(new TimerTask(){
            @Override
            public void run() {
                String id = localTaskExecuteInstance.getTaskExecuteInstanceID();
                LocalTaskLockApi lockMain = LocalTaskLockApi.getLock("localTaskLockMain" + "-id-" + id);

                if(lockMain.lockNotBlock()){ // 同时只能有一条记录
                    try{
                        LocalTaskExecuteInstance localTaskExecuteInstanceNow = localTaskExecuteInstanceDao.selectById(id);
                        if(localTaskExecuteInstanceNow.getStatus() == LocalTaskExecuteInstance.status_finish
                                || localTaskExecuteInstanceNow.getStatus() == LocalTaskExecuteInstance.status_fail){
                            return;
                        }

                        startTaskExecuteProcessorOneSync(localTaskExecute, localTaskExecuteInstance, failRetryTimeSeconds);
                    }finally{
                        lockMain.unlockIfSuccess();

                        // 移除本地缓存
                        IdTimeDTO idTimeDTO = new IdTimeDTO();
                        idTimeDTO.setId(localTaskExecuteInstance.getTaskExecuteInstanceID());
                        idTimeDTO.setActionTime(localTaskExecuteInstance.getActionTime().getTime());
                        idTimeMap.remove(idTimeDTO);
                    }
                }else{
                    log.info("已经有任务正在进行, taskExecuteInstance={}", ObjectUtil.toString(localTaskExecuteInstance));
                }
            }
        }, date);
    }

    private void startTaskExecuteProcessorOneSync(LocalTaskExecute localTaskExecute,
                                                  LocalTaskExecuteInstance localTaskExecuteInstance,
                                                  int[] failRetryTimeSeconds){
        LocalTaskExecuteInstance localTaskExecuteInstanceNext = null;
        try{
            // 执行任务
            localTaskExecuteInstanceNext = taskOneInstanceStart(localTaskExecute, localTaskExecuteInstance, failRetryTimeSeconds);
        }catch(Exception e){
            log.error("taskOneInstanceStart异常, tTaskExecuteInstanceID={}", localTaskExecuteInstance.getTaskExecuteInstanceID(), e);
        }finally{
            if(localTaskExecuteInstanceNext != null){ // 下一次触发
                if(localTaskExecuteInstanceNext.getActionTime().getTime() <= (System.currentTimeMillis() + LocalTaskCommon.taskScanTime_extend)){
                    // this.startTaskExecuteProcessor(type);
                    startTaskExecuteProcessorOne(localTaskExecuteInstanceNext, localTaskExecute, failRetryTimeSeconds);
                }
            }
        }
    }

    @SneakyThrows
    private LocalTaskExecuteInstance taskOneInstanceStart(LocalTaskExecute localTaskExecute, LocalTaskExecuteInstance localTaskExecuteInstance, int[] failRetryTimeSeconds) {
        LocalTaskExecuteInstance returnLocalTaskExecuteInstance = null;
        try{
            String executeJSON = localTaskExecuteInstance.getExecuteJSON();
            Object parse = JSONObject.parse(executeJSON);
            Object t = parse;

            LocalTask localTask = new LocalTask();
            localTask.setExecuteCount(localTaskExecuteInstance.getExecuteCount());
            localTask.setTaskExecuteInstanceID(localTaskExecuteInstance.getTaskExecuteInstanceID());
            localTask.setLocalTaskService(localTaskService);
            localTask.setActionTime(localTaskExecuteInstance.getActionTime());

            LocalTaskConsumeResult localTaskConsumeResult = new LocalTaskConsumeResult();
            localTaskConsumeResult.setResult(LocalTaskConsumeResult.result_reconsumeLaterConsumer);
            try{
                LocalTaskConsumeResult localTaskConsumeResultTem = localTaskExecute.doConsume(localTask, t);
                if(localTaskConsumeResultTem != null){
                    localTaskConsumeResult = localTaskConsumeResultTem;
                }
            }catch(Exception e){
                log.error("本地任务执行器, 一个任务执行异常 任务id={}, t={}", localTaskExecuteInstance.getTaskExecuteInstanceID(),
                        Optional.ofNullable(t).map(o -> o.getClass()).map(o-> o.getName()).orElse(null),
                        e);
            }

            if(localTaskConsumeResult.getResult() == LocalTaskConsumeResult.result_commitConsumer){
                // 任务执行成功
            }else if(localTaskConsumeResult.getResult() == LocalTaskConsumeResult.result_failConsumer){
                // 任务执行失败
                localTaskExecuteInstanceDao.updateStatusById(localTaskExecuteInstance.getTaskExecuteInstanceID(), LocalTaskExecuteInstance.status_fail);
            }else if(localTaskConsumeResult.getResult() == LocalTaskConsumeResult.result_reconsumeLaterConsumer){
                // 任务执行失败 重试任务
                if(failRetryTimeSeconds == null || failRetryTimeSeconds.length == 0){ // 没有重试机制使用默认重试机制
                    Integer executeCount = localTaskExecuteInstance.getExecuteCount();
                    if(executeCount > 3){
                        localTaskExecuteInstanceDao.updateStatusById(localTaskExecuteInstance.getTaskExecuteInstanceID(), LocalTaskExecuteInstance.status_fail);
                    }else{
                        localTaskExecuteInstanceDao.updateStatusById(localTaskExecuteInstance.getTaskExecuteInstanceID(), LocalTaskExecuteInstance.status_run);
                        localTaskExecuteInstanceDao.updateExecuteCount(localTaskExecuteInstance.getTaskExecuteInstanceID(), executeCount + 1);
                    }
                }else{
                    Integer executeCount = localTaskExecuteInstance.getExecuteCount();
                    Integer failRetryTimeSecond = Optional.ofNullable(failRetryTimeSeconds).filter(o-> o.length > executeCount).map(o-> o[executeCount]).orElse(null);
                    if(failRetryTimeSecond == null){
                        // 终止
                        localTaskExecuteInstanceDao.updateStatusById(localTaskExecuteInstance.getTaskExecuteInstanceID(), LocalTaskExecuteInstance.status_fail);
                    }else{
                        long nowActionTime = localTaskExecuteInstance.getActionTime().getTime() + failRetryTimeSecond * 1000L;
                        localTaskExecuteInstance.setActionTime(new Timestamp(nowActionTime));
                        localTaskExecuteInstance.setStatus(LocalTaskExecuteInstance.status_run);
                        localTaskExecuteInstance.setExecuteCount(executeCount + 1);
                        localTaskExecuteInstanceDao.updateById(localTaskExecuteInstance);

                        returnLocalTaskExecuteInstance = localTaskExecuteInstance;
                    }
                }

            }

            return returnLocalTaskExecuteInstance;
        }catch(Exception e){
            throw e;
        }
    }

    /**
     * 扫描可执行任务
     * @param type 任务类型
     * @param sameTimeWorkNum 每个社区同一时间执行数量
     */
    private List<LocalTaskExecuteInstance> selectTaskList(String type, Integer sameTimeWorkNum){
        LocalTaskLockApi lock = LocalTaskLockApi.getLock("loaclTaskSelectTasklock" + "-type-" + type);
        try{
            lock.lockBlock();
            return selectTaskListMain(type, sameTimeWorkNum);
        } catch (Exception e) {
            log.error("本地任务执行器扫描任务等待超时" + "-type-" + type, e);
        } finally{
            Optional.ofNullable(lock).ifPresent(o-> o.unlockIfSuccess());
        }
        return null;
    }

    @SneakyThrows
    private List<LocalTaskExecuteInstance> selectTaskListMain(String type, Integer sameTimeWorkNum) {
        try{
            long now = System.currentTimeMillis();
            List<LocalTaskExecuteInstance> localTaskExecuteInstances = localTaskExecuteInstanceDao.selectGroup(
                    LocalTaskCommon.springApplicationName,
                    LocalTaskCommon.taskSelfFlag ? LocalTaskCommon.serverId: null,
                    type,
                    sameTimeWorkNum,
                    new Timestamp(now + LocalTaskCommon.taskScanTime_extend));
            if(localTaskExecuteInstances == null || localTaskExecuteInstances.size() == 0){
                return null;
            }
            // 取全部id
            List<String> idSqls = new ArrayList<>();
            for(LocalTaskExecuteInstance oneCid : localTaskExecuteInstances){
                String[] taskExecuteInstanceIDArr = Optional.ofNullable(oneCid).map(o-> o.getTaskExecuteInstanceIDList()).map(o-> o.trim())
                        .filter(o-> !o.equals("")).map(o-> o.split(",")).orElseGet(()-> new String[0]);
                for(String id : taskExecuteInstanceIDArr){
                    idSqls.add(id);
                }
            }

            List<LocalTaskExecuteInstance> localTaskExecuteInstancesAll = new ArrayList<LocalTaskExecuteInstance>();
            if(idSqls != null && idSqls.size() != 0){
                localTaskExecuteInstancesAll = localTaskExecuteInstanceDao.selectBatchIds(idSqls);
            }

            return localTaskExecuteInstancesAll;
        }catch(Exception e){
            log.error("本地任务执行器-扫描任务异常", e);
            return null;
        }
    }

    /**
     * 发起一个任务独立事务
     */
    @SneakyThrows
    public <T> void sendTask(String type,
                              T t,
                              Timestamp actionTime,
                              Boolean async){ // , 异步标记
        long now = System.currentTimeMillis();

        String json = JSONObject.toJSONString(t, SerializerFeature.WriteClassName);
        LocalTaskExecuteInstance localTaskExecuteInstance = new LocalTaskExecuteInstance();
        localTaskExecuteInstance.setTaskExecuteInstanceID(IdUtil.getSnowflake().nextIdStr() + "");
        localTaskExecuteInstance.setStatus(LocalTaskExecuteInstance.status_await);
        localTaskExecuteInstance.setExecuteCount(0);
        localTaskExecuteInstance.setExecuteJSON(json);
        localTaskExecuteInstance.setServerName(LocalTaskCommon.springApplicationName);

        if(LocalTaskCommon.taskSelfFlag){
            localTaskExecuteInstance.setServerId(LocalTaskCommon.serverId);
        }

        localTaskExecuteInstance.setType(type);
        localTaskExecuteInstance.setActionTime(actionTime);

        localTaskExecuteInstance.setStContext(Optional.ofNullable(ThreadUtil.getThreadVariable("stContext")) // 上下文对象
                .map(o-> JSONObject.toJSONString(o, SerializerFeature.WriteClassName)).orElse(null));

        localTaskExecuteInstanceDao.insert(localTaskExecuteInstance);

        log.debug("任务插入数据 actionTime={}", actionTime.getTime());

        if(async){ // 异步执行
            // 如果可以开始则直接执行
            if(actionTime.getTime() <= (now + LocalTaskCommon.taskScanTime_extend)){

                boolean inTransaction = TransactionSynchronizationManager.isActualTransactionActive();
                //非事务状态，直接执行，不做任何保证。
                if (!inTransaction) {
                    localTaskExecuteProcessor.startTaskExecuteProcessor(type);
                }else{
                    TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
                        @SneakyThrows
                        @Override
                        public void afterCommit() {
                            //事务后执行
                            localTaskExecuteProcessor.startTaskExecuteProcessor(type);
                        }
                    });
                }
            }
        }else{ // 同步执行
            boolean inTransaction = TransactionSynchronizationManager.isActualTransactionActive();
            //非事务状态，直接执行，不做任何保证。
            if (!inTransaction) {
                Future<?> future = syncThreadPool.submit(() -> {
                    localTaskExecuteProcessor.startTaskExecuteProcessorSync(type, localTaskExecuteInstance);
                });
                future.get(60L, TimeUnit.SECONDS);
            }else{
                TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
                    @SneakyThrows
                    @Override
                    public void afterCommit() {
                        //事务后执行
                        Future<?> future = syncThreadPool.submit(() -> {
                            localTaskExecuteProcessor.startTaskExecuteProcessorSync(type, localTaskExecuteInstance);
                        });
                        future.get(60L, TimeUnit.SECONDS);
                    }
                });
            }
        }
    }

    /**
     * 删除任务 独立事务
     * @param taskExecuteInstanceID
     */
    @SneakyThrows
    public void taskDelete(String taskExecuteInstanceID) {

        try{
            long now = System.currentTimeMillis();
            LocalTaskExecuteInstance localTaskExecuteInstance = localTaskExecuteInstanceDao.selectById(taskExecuteInstanceID);
            if (localTaskExecuteInstance.getActionTime().getTime() <= now + 30L * 1000L) {
                // 任务已经开始,或马上开始, 无法更改
                log.error("任务已经开始或将要开始, 无法终止");
                throw new RuntimeException("任务已经开始无法终止");
            }
            localTaskExecuteInstanceDao.deleteById(taskExecuteInstanceID);
        }catch(Exception e){
            throw e;
        }
    }
}
